package chenxu.test.hbase.storm;

import java.io.IOException;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class ORHbaseTupleProducer implements HbaseTupleProducer {

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	@Override
	public Values toTuple(Result result) throws IOException {
		if(null==result) {
			throw new IOException("null result");
		}
		
		byte[] row = result.getRow();
		String string = Bytes.toString(row);
		String[] split = StringUtils.split(string, (char)1);
		String objectid = split[0];
		String dataid = split[1];
		
		KeyValue keyValue = result.raw()[0];
		String protocoltype = Bytes.toString(keyValue.getQualifier()) ;
		long timestamp = keyValue.getTimestamp();
		
//		keyValue.getValue();暂时不传递value
		
		return new Values(new Object[]{objectid,dataid,protocoltype,timestamp,null});
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("objectid","dataid","protocoltype","timestamp","value"));
	}

}
